<?php
use Workerman\Worker;
use Workerman\Connection\TcpConnection;
use Workerman\Connection\UdpConnection;
use Workerman\Events\EventInterface; 
use \React\Promise\Deferred;
use function \Hprose\Future\co;

class Booru 
{  
    public function __debugInfo(){
        return [
            'reg_protocol'=>    self::$reg_protocol,
            'reg_addr'=>        self::$reg_addr,
            'reg_conn'=>        self::$reg_conn,
            'reg_nodes'=>       self::$reg_nodes,
            'reg_relation'=>    self::$reg_relation,
            '----------------------------------------------'=>'',
            'client_addr'=>     self::$client_addr,
            'client_tick'=>     self::$client_tick, 
            'client_nodes'=>    self::$client_nodes, 
            'client_defer'=>  self::$client_defer,
            'client_cache'=>  self::$client_cache,
        ];
    }
    ////////////// basic /////////////////

    static $reg_protocol = '\Protocols\Frame';
    static $reg_addr = '127.0.0.1:55001';
    static $reg_conn;//进程唯一 tcp接收 udp接收  
    static $reg_nodes = [];//[  TcpConnection ,  TcpConnection   ]
    static $reg_relation = [];//['coid'=>['coid'=>'true']]
 
    public static function connectServer($group = 'worker')
    {
        if (isset(self::$client_addr)) {
            return self::$client_addr;
        }
        //尝试链接 TCP
        $reg_socket = stream_socket_client('tcp://'.self::$reg_addr, $errno, $errstr) or exit;
        $addr = stream_socket_get_name($reg_socket, false); 
        //链接 TCP 成功后启动事件循环
        $event = Proc::startEventLoop();
        self::$reg_conn = new class($reg_socket, $addr) extends TcpConnection{
            public function __debugInfo(){
                return ['TcpConnection'=>$this->_remoteAddress];
            }
        };
        self::$reg_conn->protocol = self::$reg_protocol;
        self::$reg_conn->onMessage= [self::class,"onClientTcpMsg"];
        self::$reg_conn->onClose= [self::class,"onClientTcpClose"];
        self::$reg_conn->onError= [self::class,"onClientTcpClose"];
        self::$reg_conn->send(['ongroup',$group]); 
        //监听 UDP 
        $msg_conn = new \Connection\UdpListenConnection($addr);
        $msg_conn->protocol = self::$reg_protocol;
        $msg_conn->onMessage = [self::class,"onClientUdpMsg"]; 
        self::$client_addr = $addr;
        self::$client_nodes = [  self::$reg_addr=>'true'  ];//client table 
        self::$client_tick = 0;  
        return $addr;
    }

    public static function startServer()
    { 
        if (Proc::checkAddr('tcp://'.self::$reg_addr)) {
            exit("server started\n");
        } 
        //尝试监听 TCP
        $socket = \stream_socket_server('tcp://'.self::$reg_addr, $errno, $errstr) or exit; 
        //监听 TCP 成功后启动事件循环
        $event = Proc::startEventLoop();
        $event->add(1, EventInterface::EV_TIMER,[self::class,'startServerTimer']); 
        $event->add($socket, EventInterface::EV_READ, function ($socket) { 
            $socket = \stream_socket_accept($socket, 0, $addr);
            $conn = new class($socket, $addr) extends TcpConnection implements \JsonSerializable { 
                public function getAddress(){
                    return $this->_remoteAddress;
                }
                public function __debugInfo(){
                    return ['TcpConnection'=>$this->_remoteAddress];
                }
                public function jsonSerialize(){
                    return "TcpConnection";
                }
            };
            $conn->protocol  = self::$reg_protocol;
            $conn->onMessage = [self::class,"onServerTcpMsg"];
            $conn->onClose   = [self::class,"onServerTcpClose"]; 
             
            // echo ">conn:".$conn->getRemotePort()."\n";   
            // self::$workers[$addr] = $conn; 
            self::$reg_nodes[$addr] = $conn;
            self::$client_nodes[$addr] = '';
        });  
        //兼容 connectServer  
        $msg_conn = new \Connection\UdpListenConnection(self::$reg_addr);
        $msg_conn->protocol = self::$reg_protocol;
        $msg_conn->onMessage = [self::class,"onClientUdpMsg"];
        self::$client_nodes = [ self::$reg_addr=>'true' ];//client table 
        self::$client_addr = self::$reg_addr;
        self::$client_tick = 0;
        self::$reg_conn = new class{
            public function send($data){ 
                Booru::onServerTcpMsg(null,$data);
            }     
            public function getAddress(){
                return Booru::$reg_addr;
            }
        };  
        self::$reg_nodes = [ self::$reg_addr=>self::$reg_conn  ];
        //子进程
        echo "[START]:".'tcp://'.self::$reg_addr."\n"; 
        \Proc::fork(function () { 
            \Booru::connectServer('worker'); 
        });    
        //usleep(1000*500);//强行等待子进程
    }

    public static function startServerTimer(){ 
        foreach (self::$reg_nodes as $conn) {  
            $nodes[$conn->getAddress()] = $conn->type??''; 
            if(isset($conn->type)&&$conn->type=='worker'){
                $w[] = $conn->getAddress();
            }
        } 
        // if (count($workers)<1) { 
        //     //创建最少一个节点进程
        //     \Proc::fork(function () { 
        //         \Booru::connectServer('worker'); 
        //     });   
        // }     
        
        //同步self::$client_data   
        self::$client_nodes = $nodes;
        foreach (self::$reg_nodes as $addr => $conn) { 
            $conn->send(['sync',$nodes,$w]);
        }  
 
        //echo "[..";
        // unset(self::$client_cache);
        // file_put_contents('nil',json_encode([
        //     'time'=>microtime(true),
        //     'client_data'=>self::$client_data,    
        //     'reg_conn'=>self::$reg_conn,
        //     'nodes'=>self::$reg_nodes,
        //     'relation'=>self::$reg_relation,
        //     'local_gen'=>\lib\BaseActor::$gens,
        //     'local_id'=>\lib\BaseActor::$local_id,
        //     'timer'=>Worker::$globalEvent->_eventTimer??
        //              Worker::$globalEvent->_task??
        //              Worker::$globalEvent->_timerIdMap??
        //              [], 
        // ],128));  
        //echo ".]"; 
        // //\gc_collect_cycles(); 
    }

    /////////////// server trigger //////////////////

    // static $servers = [];//['addr'=>TcpConnection]
    // static $workers = []; //['addr'=>TcpConnection]
    static function onServerTcpMsg($conn, $data)
    {
        //echo json_encode($data)."\n";
        switch($data[0]){
            case 'spawn_opt':
                break;
            case 'ongroup':
                $conn->type = $data[1]; 
                break;
            case 'onmonitor'://0:TOKEN 1:DIE 2:NOTIFY
                self::$reg_relation[$data[1]][$data[2]]=true;
                break; 
            case 'ondemonitor'://0:TOKEN 1:DIE 2:NOTIFY
                unset(self::$reg_relation[$data[1]][$data[2]]);//=true;
                break; 
            case 'onexit'://0:TOKEN 1:COID
                unset(self::$reg_relation[$data[1]]);
                break;
            case 'ping':
                break;
            default:
                break;
        }
        //echo json_encode($data)."\n";
    }
    
    static function onServerTcpClose($conn)
    { 
        foreach (self::$reg_relation as $coid => $links) { 
            preg_match('/([\d\.]+:\d+)#(\w+)/',$coid,$arr);
            if($arr[1]==$conn->getAddress()){
                unset(self::$reg_relation[$coid]);
                foreach ($links as $id => $_) {   
                    preg_match('/([\d\.]+:\d+)#(\w+)/',$id,$arr);
                    Booru::clientCase($arr[1],['exit',$id,$coid]);//token,coid,source
                }
            }
        }
        unset(self::$reg_nodes[$conn->getAddress()]);  
    } 

    /////////////  client method ////////////// 

    static $client_addr;
    static $client_tick = 0;
    static $client_nodes = [];//[node=>type]
    static $client_defer = [];//[promise,...]
    static $client_cache = [];//[socket,....]
    static $client_workers =[];
    static function channel(&$addr,&$tick){
        $addr = self::$client_addr;
        $tick = ++self::$client_tick;//
        self::$client_defer[self::$client_tick] = //new Deferred;
        new class extends Deferred { function __debugInfo(){ return ['Deferred'=>__FILE__]; } };
        return self::$client_defer[self::$client_tick]->promise();  
        return $tick;
    }
    public static function clientCall($addr, $data)
    {   
        $defer = self::channel($data[],$data[]);
        self::clientCase($addr, $data);
        return $defer;
    }
    
    public static function clientCase($addr, $data)
    {
        if(!isset(self::$client_nodes[$addr])){ 
            print_r([$addr,self::$client_nodes,$data]);
            print_r(new self);
            throw new \Exception("Error Processing Request", 1); 
        }
        if(empty(self::$client_cache[$addr])){
            self::$client_cache[$addr] = stream_socket_client("udp://$addr", $errno, $errstr);
        }
        \fwrite(self::$client_cache[$addr] , self::$reg_protocol::encode($data)); 
    }

    public static function serverCall($data)
    {  
        $defer = self::channel($data[],$data[]); 
        self::$reg_conn->send($data); 
        return $defer;
    } 

    public static function serverCase($data){
        self::$reg_conn->send($data);         
    }

    
    /////////////  client trigger ////////////// 
    
    static function onClientUdpMsg($conn, $data)
    {
        //echo "onClientUdpMsg:".json_encode($data)."\n"; 
        switch ($data[0]) {
            case 'spawn_opt'://0:TOKEN 1:FN 2:ARGS 3:OPT 4:RETURN_IP 5:RETURN_TICK 
                //unset($data[3]['ip']);//BUG
                $coid = \lib\BaseActor::spawn_opt($data[1],$data[2],$data[3]); 
                self::clientCase($data[4], ['resolve',$data[5],$coid]);  
                break;
            case 'send': //0:TOKEN 1:COID 2:MSG
                \lib\BaseActor::send($data[1], $data[2]);
                break;
            case 'exit'://0:TOKEN 1:COID 2:REASON 
                \lib\BaseActor::exit($data[1],$data[2]); 
                break;
            case 'resolve'://0:TOKEN 1:TICK 2:RETURN_VALUE  
                if (isset( self::$client_defer[$data[1]] )) {
                    self::$client_defer[$data[1]]->resolve($data[2]);
                    unset(self::$client_defer[$data[1]]);
                } else {   
                    throw new \Exception("Error Processing Request", 1);
                } 
                break; 
            case 'reject'://0:TOKEN 1:TICK 2:RETURN_VALUE  
                if (isset( self::$client_defer[$data[1]] )) {
                    $error = new Exception($data[2], 1); 
                    self::$client_defer[$data[1]]->reject($error);
                    unset(self::$client_defer[$data[1]]);
                } else {   
                    throw new \Exception("Error Processing Request", 1);
                } 
                break;  
            case 'monitor'://0:TOKEN 1:DEATH 2:NOTIFY  3:RETURN_IP 4:RETURN_TICK  
                if( Actor::ref($data[1],$data[2]) ){   
                    self::serverCase(['onmonitor',$data[1],$data[2]]);
                    self::clientCase($data[3],['resolve',$data[4],'ok']); 
                }else{
                    self::clientCase($data[3],['reject',$data[4],'err']);   
                } 
                break;
            case 'demonitor'://0:TOKEN 1:DEATH 2:NOTIFY  3:RETURN_IP 4:RETURN_TICK  
                Actor::out($data[1],$data[2]);
                self::serverCase(['ondemonitor',$data[1],$data[2]]);
                self::clientCase($data[3],['resolve',$data[4],'ok']);
                //unset(self::$client_monitor[$data[1]][$data[2]]); 
                //self::serverCase(['demonitor',$data[1],$data[2]]); 
                //self::clientCase($data[3],['return',$data[4],'ok']);  
                break;
            default:
                
                # code...
                break;
        }
    }

    static function onClientTcpMsg($conn,$data)
    { 
        switch ($data[0]) {
            case 'sync':  //同步self::$client_data
                self::$client_nodes = $data[1];
                self::$client_workers = $data[2]; 
                break; 
            default:
                # code...
                break;
        } 
    }
    
    static function onClientTcpClose($conn)
    { 
        exit;//失去链接时候死亡
    }

    
}